package api

import (
	"context"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/jinzhu/copier"
	"github.com/pkg/errors"
	"github.com/zeromicro/go-zero/core/logx"
	"go-zero-demo/common/global"
	"go-zero-demo/common/tools"
	"go-zero-demo/common/tools/limit"
	"go-zero-demo/user-api/internal/svc"
	"go-zero-demo/user-api/internal/types"
	"go-zero-demo/user-rpc/pb"
	"math/rand"
	"strconv"
)

type UserInfoLogic struct {
	logx.Logger
	ctx    context.Context
	svcCtx *svc.ServiceContext
}

func NewUserInfoLogic(ctx context.Context, svcCtx *svc.ServiceContext) *UserInfoLogic {
	return &UserInfoLogic{
		Logger: logx.WithContext(ctx),
		ctx:    ctx,
		svcCtx: svcCtx,
	}
}

func (l *UserInfoLogic) UserInfo(req *types.UserInfoReq) (resp *types.UserInfoResp, err error) {
	logx.Infof("user info request:%s", tools.BeanToJson(req))
	//passBool, err2 := l.periodLimitVerify("userInfo")
	passBool, err2 := l.tokenLimitVerify("userInfo")
	if !passBool {
		logx.Error(err2)
		return nil, errors.New("您的操作频次太快,请求稍后重试!")
	}
	//err3 := l.sendSimpleMsg(err)
	//err3 := l.sendDelayMsg(err)
	//err3 := l.sendOrderMsg()
	err3 := l.sendTransactionMsg(err)
	if err3 != nil {
		return nil, err3
	}
	if err2 != nil {
		return nil, err2
	}
	userInfo, err := l.svcCtx.UserCenterClient.UserInfo(l.ctx, &pb.UserInfoReq{UserId: req.UserId})
	logx.Info("userInfo:", userInfo)
	returnObj := new(types.UserInfoResp)
	_ = copier.Copy(returnObj, userInfo)
	//fmt.Println(l.svcCtx.DistributedIdClient.GetSnowId(l.ctx, &distributed.EmptyReq{}))
	//fmt.Println(l.svcCtx.DistributedIdClient.GetId(l.ctx, &distributed.IdReq{
	//	Tag: "distributed",
	//}))
	return returnObj, nil
}

// sendOrderMsg 发送顺序消息
func (l *UserInfoLogic) sendOrderMsg() error {
	var shardingKey = "123456"
	for i := 0; i < 6; i++ {
		msg := &primitive.Message{
			Topic: "user-api-register-topic",
			Body:  []byte("Hello RocketMQ Go Client! " + strconv.Itoa(i+1)),
		}
		// Tag 设置
		msg.WithTag("user-api")
		// 设置路由的key 顺序消息中用来区分 Topic 中不同分区的关键字段，
		// 和普通消息的 Key 是完全不同的概念。消息队列 RocketMQ
		// 会将设置了相同 Sharding Key 的消息路由到同一个分区下，同一个分区内的消息将按照消息发布顺序进行消费
		msg.WithShardingKey(shardingKey)
		syncRes, err := l.svcCtx.UniversalProducer.SendSync(l.ctx, msg)
		if err != nil {
			logx.Errorf("UniversalProducer#SendSync,fail:%s", err.Error())
			return err
		}
		logx.Info(syncRes)
	}
	return nil
}

// sendTransactionMsg 发送普通消息
func (l *UserInfoLogic) sendTransactionMsg(err error) error {
	msg := &primitive.Message{
		Topic: "user-api-register-topic",
		Body:  []byte("Hello RocketMQ This TransactionMsg! " + strconv.Itoa(rand.Int())),
	}
	// Tag 设置
	msg.WithTag("user-api")
	syncRes, err := l.svcCtx.TransactionProducer.SendMessageInTransaction(l.ctx, msg)
	if err != nil {
		logx.Errorf("UniversalProducer#SendSync,fail:%s", err.Error())
		return err
	}
	logx.Info(syncRes)
	return nil
}

// sendSimpleMsg 发送普通消息
func (l *UserInfoLogic) sendDelayMsg(err error) error {
	msg := &primitive.Message{
		Topic: "user-api-register-topic",
		Body:  []byte("Hello RocketMQ Go Client! " + strconv.Itoa(rand.Int())),
	}
	// Tag 设置
	msg.WithTag("user-api")
	// 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 设置延迟等级
	msg.WithDelayTimeLevel(3)
	syncRes, err := l.svcCtx.UniversalProducer.SendSync(l.ctx, msg)
	if err != nil {
		logx.Errorf("UniversalProducer#SendSync,fail:%s", err.Error())
		return err
	}
	logx.Info(syncRes)
	return nil
}

// sendSimpleMsg 发送普通消息
func (l *UserInfoLogic) sendSimpleMsg(err error) error {
	msg := &primitive.Message{
		Topic: "user-api-register-topic",
		Body:  []byte("Hello RocketMQ Go Client! " + strconv.Itoa(rand.Int())),
	}
	// Tag 设置
	msg.WithTag("user-api")
	syncRes, err := l.svcCtx.UniversalProducer.SendSync(l.ctx, msg)
	if err != nil {
		logx.Errorf("UniversalProducer#SendSync,fail:%s", err.Error())
		return err
	}
	logx.Info(syncRes)
	return nil
}

// tokenLimitVerify  限流 令牌桶算法实现
func (l *UserInfoLogic) tokenLimitVerify(redisKey string) (bool, error) {
	tokenLimiter := limit.NewTokenLimiter(1, 1, l.svcCtx.RedisClient, global.TokenLimitPrefix+redisKey)
	return tokenLimiter.AllowCtx(l.ctx), nil
}

// periodLimitVerify 限流 计数器 滑动窗口
func (l *UserInfoLogic) periodLimitVerify(redisKey string) (bool, error) {
	// period 窗口大小，单位s
	// quota 请求上限
	// align 线性限流，开启此选项后可以实现周期性的限流 比如quota=5时，quota实际值可能会是5.4.3.2.1呈现出周期性变化
	periodLimit := limit.NewPeriodLimit(10, 1, l.svcCtx.RedisClient, global.PeriodLimitPrefix)
	// code source
	takeCode, err := periodLimit.TakeCtx(l.ctx, redisKey)
	if err != nil {
		logx.Error(err)
		return true, err
	}
	switch takeCode {
	case limit.OverQuota:
		logx.Errorf("OverQuota key: %v", redisKey)
		return false, errors.New("OverQuota key:" + redisKey)
	case limit.Allowed:
		fallthrough
	case limit.HitQuota:
		logx.Infof("AllowedQuota key: %v", redisKey)
		return true, nil
	default:
		logx.Errorf("DefaultQuota key: %v", redisKey)
		// unknown response, we just let the sms go
		return false, errors.New("DefaultQuota key:" + redisKey)
	}
}
